// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once
#include <ctime>  // strptime
#include <experimental/filesystem>
#include <memory>
#include <regex>
#include <string>
#include <utility>
#include <vector>

#include <boost/asio.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <fmt/format.h>

#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "orio/io/base.h"
#include "orio/io/concat.h"
#include "orio/record/topic.h"

class FlowWriter : orio::io::Writer<orio::record::TopicRecord> {
 public:
  FlowWriter(
      const std::string& root_dir,  // may or may not end with '/'
      const std::string& machine,
      const std::string& topic_string,
      const std::string& extension = ".gz")
      : root_(root_dir), machine_(machine), topic_string_(topic_string), extension_(extension) {
    CHECK_THROW(extension.empty() || extension == ".gz" || extension == ".lz4")
        << "[FlowWriter] Unsupported compression type: " + extension;
  }

  FlowWriter(
      boost::asio::io_context::strand* strand,
      const std::string& root_dir,  // may or may not end with '/'
      const std::string& machine,
      const std::string& topic_string,
      const std::string& extension = ".gz")
      : FlowWriter(root_dir, machine, topic_string, extension) {
    strand_ = strand;
    expire_timer_ = std::make_unique<boost::asio::deadline_timer>(strand_->context());
  }

  void Write(const orio::record::TopicRecord& in) override {
    CheckOrResetFile(in.timestamp);
    writer_->Write(in);
  }

  void Flush() override { writer_->Flush(); }

 private:
  void TimerHandler(int64_t now, const boost::system::error_code& error) {
    // timer is cancel when FlowWriter is destroyed
    if (error == boost::asio::error::operation_aborted) return;

    LOG(INFO) << "[FlowWriter] timer check " << last_tmp_;
    CheckOrResetFile(now);
  }

  void CheckOrResetFile(int64_t now) {
    if (!writer_ || now >= expire_timestamp_) {
      // move to the next file
      auto t = PtimeFromTimestamp(now);

      unsigned int seed = now;

      std::string queue(10, '0');
      if (topic_string_.back() == '2' || *(topic_string_.rbegin() + 1) == '2') {
        queue[2] = '3';  // 0000 => 0030
      }

      queue[4] = rand_r(&seed) % 26 + 'a';
      for (int i = 5; i < 10; i++) {
        queue[i] = rand_r(&seed) % 36;
        if (queue[i] < 10) {
          queue[i] += '0';
        } else {
          queue[i] += 'a' - 10;
        }
      }

      auto yyyymmdd = boost::gregorian::to_iso_string(t.date());
      auto hhmmss = boost::posix_time::to_iso_string(t.time_of_day()).substr(0, 6);
      std::experimental::filesystem::path path = root_;
      path /= machine_;
      path /= yyyymmdd;
      std::string topic_last3byte = topic_string_.substr(topic_string_.size()-3);
      static const std::regex group_worker_regex(R"(_[a-p][12])");
      std::smatch m;
      if (std::regex_search(topic_string_, m, group_worker_regex)) {
        path /= topic_string_.substr(0, topic_string_.size()-3);
      }
      std::string filename = fmt::format(
          "{}.queue-{}.{}-{}Z{}",  // filename
          topic_string_,                // includes worker_id
          queue,                        // random generated
          yyyymmdd,                     // date
          hhmmss,                       // ignore nano
          extension_);                  // gz
      auto tmp = path;
      tmp /= "_" + filename;
      path /= filename;

      // expire_timestamp_ = (now / 86400'000'000'000ll + 1) * 86400'000'000'000ll;
      expire_timestamp_ = (now / 1800'000'000'000ll + 1) * 1800'000'000'000ll;
      LOG(INFO) << "[FlowWriter] " << path;
      LOG(INFO) << "[FlowWriter] " << now << " " << expire_timestamp_;

      writer_.reset(orio::record::TopicWriter::ToFile(tmp));

      if (!last_tmp_.empty()) {
        // move _file.gz to file.gz
        std::rename(last_tmp_.c_str(), last_path_.c_str());
      }
      last_path_ = std::move(path);
      last_tmp_ = std::move(tmp);

      if (expire_timer_) {
        expire_timer_->expires_at(t);
        expire_timer_->async_wait(
            strand_->wrap(std::bind(
                &FlowWriter::TimerHandler,
                this,
                expire_timestamp_,
                std::placeholders::_1)));
      }
    }
  }
  std::string last_path_;
  std::string last_tmp_;
  std::string root_;
  std::string machine_;
  std::string topic_string_;
  std::string extension_;
  int64_t expire_timestamp_ = 0;
  std::unique_ptr<orio::io::Writer<orio::record::TopicRecord>> writer_;
  std::unique_ptr<boost::asio::deadline_timer> expire_timer_;
  boost::asio::io_context::strand* strand_ = nullptr;
};
